package cc.toprank.syncdatatask.listener;

import cc.toprank.syncdatatask.pojo.MsgInfo;
import com.alibaba.druid.util.StringUtils;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.toprank.druid.store.jdbc.SqlTemplate;
import com.toprank.druid.store.jdbc.builder.DeleteSql;
import com.toprank.druid.store.jdbc.builder.InsertSql;
import com.toprank.druid.store.jdbc.builder.SelectSql;
import com.toprank.druid.store.jdbc.builder.UpdateSql;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;

/**
 * description 接收离线数据并注入sql
 * User: WeiDa
 * Date: 2017-5-7
 * Time: 16:57
 */
public class QueueListener1 {

    @Autowired
    SqlTemplate sqlTemplate;

    private Logger logger = Logger.getLogger(QueueListener1.class);

    public void displayMsgInfo(MsgInfo msgInfo) throws Exception {
        try {
            if (msgInfo == null) return;
            if (msgInfo.isFlag()) return;
            msgInfoHandler(msgInfo);
        } catch (Exception e) {
            e.printStackTrace();
            logger.error(e.getCause());
        } finally {
            return;
        }
    }

    private void msgInfoHandler(MsgInfo msgInfo) {

        List<CanalEntry.Column> data = msgInfo.getColumnList();
        logger.debug("服务端接收到来自客户端的消息【'" + msgInfo.getMsgId() + "】成功");

        if (msgInfo.getEventType() == CanalEntry.EventType.INSERT) {
            //todo save
            insert(msgInfo.getTableName(), data);
            logger.info("save result:=====success");
        }else if (msgInfo.getEventType() == CanalEntry.EventType.UPDATE) {
            //todo update
            update(msgInfo.getTableName(), msgInfo.getPrimaryKey(), msgInfo.getIdValue(), data);
        }else if (msgInfo.getEventType() == CanalEntry.EventType.DELETE) {
            //todo delete
            delete(msgInfo.getTableName(), msgInfo.getPrimaryKey(), msgInfo.getIdValue());
        }

    }

    /**
     * dazi
     *
     * @param tableName
     * @param primaryKey
     * @param idValue
     * @return
     */
    private int select(String tableName, String primaryKey, Object idValue) {
        SelectSql selectSql = new SelectSql(sqlTemplate);
        return selectSql.select()
                .columns("count(1)")
                .from()
                .table(tableName)
                .where(primaryKey + " = ?", idValue)
                .single();
    }

    /**
     * description
     * User: WeiDa
     * Date: 2017-3-23
     * Time: 14:00
     */
    private int update(String tableName, String primaryKey, Object idValue, List<CanalEntry.Column> columnList) {


        try {
            String syncTime = "",syncTimeResult = "";
            UpdateSql updateSql = new UpdateSql(sqlTemplate);
            updateSql.update()
                    .table(tableName);
            for (CanalEntry.Column column : columnList) {

                if(column.getName().equalsIgnoreCase("SyncTime")){
                    syncTime = column.getValue();
                    continue;
                }
                if(column.getName().equalsIgnoreCase("SyncTimeResult")){
                    syncTimeResult = column.getValue();
                    continue;
                }
                if (column.getUpdated()) {
                    if (StringUtils.isEmpty(column.getValue())) {
                        updateSql.set(column.getName(), null);
                    }else{
                        if (column.getMysqlType().equalsIgnoreCase("bit(1)")) {
                            updateSql.setOnNotNull(column.getName(), Integer.parseInt(column.getValue()));
                        }else{
                            updateSql.setOnNotNull(column.getName(), column.getValue());
                        }
                    }
                }
            }
            if(syncTime.equalsIgnoreCase(syncTimeResult)) return 0;
            LocalDateTime date = LocalDateTime.now();
            updateSql.setOnNotNull("SyncTime", date);
            updateSql.setOnNotNull("SyncTimeResult", date);
            logger.info("update result:=====success idValue"+idValue);
            return updateSql.where(primaryKey + " = ?", idValue)
                    .doUpdate();
        } catch (Exception e) {
            e.printStackTrace();
            logger.error(e.getCause());
            logger.error(e.getLocalizedMessage());
        } finally {
            return 0;
        }
    }

    /**
     * description
     * User: WeiDa
     * Date: 2017-3-23
     * Time: 14:01
     */
    private void insert(String table, List<CanalEntry.Column> columnList) {

        try {
            String syncTime = "",syncTimeResult = "";
            List<String> keyList = new ArrayList<>();
            List<Object> valueList = new ArrayList<>();
            for (CanalEntry.Column column : columnList) {
                if (StringUtils.isEmpty(column.getValue())) continue;
                if (column.getName().equalsIgnoreCase("IsSync")) continue;
                if(column.getName().equalsIgnoreCase("SyncTime")){
                    syncTime = column.getValue();
                    continue;
                }
                if(column.getName().equalsIgnoreCase("SyncTimeResult")){
                    syncTimeResult = column.getValue();
                    continue;
                }
                keyList.add(column.getName());
                if (column.getMysqlType().equalsIgnoreCase("bit(1)")) {
                    valueList.add(Integer.parseInt(column.getValue()));
                } else if (column.getMysqlType().equalsIgnoreCase("int(11)") && StringUtils.isEmpty(column.getValue())) {
                    valueList.add(0);
                } else if (column.getMysqlType().equalsIgnoreCase("bigint(20)") && StringUtils.isEmpty(column.getValue())) {
                    valueList.add(0);
                }else {
                    valueList.add(column.getValue());
                }
            }

            if(syncTime.equalsIgnoreCase(syncTimeResult)) return;
            LocalDateTime date = LocalDateTime.now();
            keyList.add("SyncTime");
            keyList.add("SyncTimeResult");
            valueList.add(date);
            valueList.add(date);
            String[] str = new String[keyList.size()];
            InsertSql insertSql = new InsertSql(sqlTemplate)
                    .insert(table)
                    .columns(keyList.toArray(str)
                    );
            insertSql.values(valueList.toArray()
            );
            insertSql.doBatchInsert();
        } catch (Exception e) {
            e.printStackTrace();
            logger.error(e.getCause());
            logger.error(e.getLocalizedMessage());
        }finally {
            return;
        }
    }



    private int delete(String tableName, String primaryKey, Object idValue){

        try {
            DeleteSql deleteSql = new DeleteSql(sqlTemplate)
                    .delete()
                    .from()
                    .table(tableName)
                    .where(primaryKey+ " = ?",idValue);
            return deleteSql
                    .doDelete();
        } catch (Exception e) {
            e.printStackTrace();
            logger.error(e.getCause());
            logger.error(e.getLocalizedMessage());
        }finally {
            return 0;
        }
    }

    protected void printColumn(List<CanalEntry.Column> columns) {
        for (CanalEntry.Column column : columns) {
            StringBuilder builder = new StringBuilder();
            builder.append(column.getName() + " : " + column.getValue());
            builder.append("    type=" + column.getMysqlType());
            if (column.getUpdated()) {
                builder.append("    update=" + column.getUpdated());
            }

            logger.info(builder.toString());
        }
    }
}
